package com.lz.springboot.rabbit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import com.lz.springboot.rabbit.common.RabbitConstants;


@Configuration
@ConfigurationProperties(prefix = "spring.data.rabbit")
public class FirstRabbitConfig{
	
	private static final Logger LOGGER = LoggerFactory.getLogger(FirstRabbitConfig.class);

	@Value("${spring.rabbitmq.host:127.0.0.1}")
	private String host;
	@Value("${spring.rabbitmq.port:5672}")
	private Integer port;
	@Value("${spring.rabbitmq.username:rabbitmq}")
	private String username;
	@Value("${spring.rabbitmq.password:rabbitmq}")
	private String password;
	@Value("${spring.rabbitmq.virtual-host:lzmh}")
	private String virtualHost;
	@Value("${spring.rabbitmq.publisher-confirms:false}")
	private Boolean publisherConfirms;
	@Value("${spring.rabbitmq.publisher-returns:false}")
	private Boolean publisherReturns;
    
	@Bean(name=RabbitConstants.FIRST_RABBIT_CONNECTION_FACTORY)
    @Primary
    public ConnectionFactory firstRabbitConnectionFactory(){
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setVirtualHost(virtualHost);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setPublisherConfirms(publisherConfirms);
        connectionFactory.setPublisherReturns(publisherReturns);
        return connectionFactory;
    }
	
	@Bean(name=RabbitConstants.FIRST_RABBIT_TEMPLATE)
    @Primary
    public RabbitTemplate firstRabbitTemplate(@Qualifier(RabbitConstants.FIRST_RABBIT_CONNECTION_FACTORY) ConnectionFactory connectionFactory){
        RabbitTemplate firstRabbitTemplate = new RabbitTemplate(connectionFactory);
        
        //消息发送到RabbitMQ交换器后接收ack回调,必须设置publisherConfirms=true
        firstRabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
			@Override
			public void confirm(CorrelationData correlationData, boolean ack, String cause) {
				if (ack) {
					if(correlationData != null) {
						LOGGER.info("message send exchange success, id={}", correlationData.getId());
					}else {
						LOGGER.info("message send exchange success");
					}
				} else {
					LOGGER.warn("message send exchange fail, cause={}", cause);
				}
			}
		});
        
        //消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调,必须设置publisherReturns=true
        firstRabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
			@Override
			public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
				 String correlationId = message.getMessageProperties().getCorrelationIdString();
				 LOGGER.warn("message send fail correlationId={} replyCode={} replyText={} exchange={}  routingKey={}", 
						 correlationId, replyCode, replyText, exchange, routingKey);
			}
		});
        return firstRabbitTemplate;
    }
	
	@Bean(name=RabbitConstants.FIRST_RABBIT_CONTAINER_FACTORY)
	@Primary
    public SimpleRabbitListenerContainerFactory firstRabbitContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,
    		@Qualifier(RabbitConstants.FIRST_RABBIT_CONNECTION_FACTORY) ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        //指定最小的消费者数量
        factory.setConcurrentConsumers(1);
        //指定最大的消费者数量
        factory.setMaxConcurrentConsumers(5);
        //指定一个请求能处理多少个消息
        factory.setPrefetchCount(10);
        configurer.configure(factory, connectionFactory);
        return factory;
    }
}